 
/* 
 * Copyright (C) Ignacio Lucas
 */
 
/*
 *  jobs_controller, hecho por Ignacio Lucas
 *
 *  Dispatcher de jobs listos para ser ejecutados. Dichos jobs pueden estar relacionados con jobs nuevos o jobs
 *  que, luego de haberse bloqueado, vuelven a estar listos para ser ejecutados.
 *  Recibe los eventos generados por el modulo io_handler y actua en consecuencia:
 *	- acepta conexiones (si el evento indica que se recibio una nueva conexion), creando una nueva tarea asociada.
 *	- recibe eventos sobre notificaciones de operaciones de E/S listas para ejecutarse.
 *	- encola nuevas tareas o tareas previamente suspendidas en una cola de tareas, de la que el grupo de server_worker consume
 *    para procesar.
 *	
 *
 *  Ignacio Lucas <ignacio.lucas@gmail.com>
 */
#include "common.h"
#include "jobs_controller.h"
#include "priority_queue.h"
#include "worker.h"
#include "server_worker.h"
#include "io_handler.h"
#include "server_worker_implementations.h"
#include "common_objects_pool.h"
#include "utils/hashmap.h"
#include "timer_service.h"
#include "conf/conf_file.h"

#include <sys/types.h>
#include <dlfcn.h>

// este numero debe ser mayor a 1.
#define MAX_PENDING_JOBS_COMMIT 32

/* 
estructura para resguardar jobs listos para ser comiteados a la cola todos juntos.
Esto permite acceder una sola vez a la cola de tareas y enviarle en conjunto hasta 
MAX_PENDING_JOBS_COMMIT tareas juntas.
*/
typedef struct s_pending_jobs
{
	server_worker_job_t worker_jobs[MAX_PENDING_JOBS_COMMIT];
	int count;

} pending_jobs;

// modulo cargado dinamicamente que provee la implementacion de mashup-server.
typedef	struct s_mashup_server_impl
{
	void *lib_handle;
	server_worker_implementation_init_module fn_app_init;

	server_worker_implementation_destroy_module fn_app_destroy;
	// funcion en la que se delega el procesamiento de las tareas. Representa la aplicacion del usuario (ver "server_worker_implementations.h").
	server_worker_implementation_app_module fn_app_implementation;

}mashup_server_impl;


typedef struct s_jobs_controller
{
	worker_t worker;  // worker asociado a jobs_controller (hilo de ejecucion donde jobs_controller ejecuta).
	server_worker_t *server_workers;  // grupo de server_workers (hilos de ejecucion que procesan las tareas).
	priority_queue_t q;  // cola de tareas.
	io_handler_t ioh;  // modulo de notificacion de eventos.
	mashup_server_core_config_t core_config; // configuracion de la aplicacion
	pending_jobs p_jobs;// jobs pendientes de comitear a la cola de tareas.
	mashup_server_impl impl;//implementacion de mashup_server.

} jobs_controller;

// Singleton de jobs_controller
static jobs_controller main_jobs_controller_instance;
static jobs_controller *main_jobs_controller = &main_jobs_controller_instance;

// macro para acceder al endpoint de un servidor dado un job_event que sabemos es de servidor.
#define jobs_controller_get_server_endpoint(event)	((mashup_server_endpoint_t *)( ((void*)event) - offsetof(mashup_server_endpoint_t,server_event_config)))

int jobs_controller_get_application_function(void **fn, const char *fn_name)
{
	int ret=0;
	dlerror();
	logger_core_info("[jobs_controller] cargando funcion: %s",fn_name);
	*(void **)(fn)=dlsym(main_jobs_controller->impl.lib_handle, fn_name);
	
	const char* dlsym_error = dlerror();
	if (dlsym_error) 
	{
		logger_core_error("[jobs_controller] error cargando funcion %s", fn_name);
		ret=1;
	}
	else
	{
		logger_core_info("[jobs_controller] funcion cargada: %s", fn_name);
	}
	return ret;
}

static void jobs_controller_endpoints_destroy()
{
	int i=0;
	for (i=0; i<main_jobs_controller->core_config.server_configs.count;i++)
	{
		mashup_server_endpoint_t *endpoint = &(main_jobs_controller->core_config.server_configs.el[i]);
		if (endpoint->server_socket != NULL)
		{
			socket_destroy(endpoint->server_socket);
		}
	}
}

static void jobs_controller_info_priority()
{
	logger_core_info("[jobs_controller] priority high. Porcentaje: %d",main_jobs_controller->core_config.q_conf.porcentaje_despacho_high);
	logger_core_info("[jobs_controller] priority normal. Porcentaje: %d",main_jobs_controller->core_config.q_conf.porcentaje_despacho_normal);
}

static int jobs_controller_start_listeners()
{
	int i=0,ret=0;
	logger_core_info("[jobs_controller] configurando %d endpoints",main_jobs_controller->core_config.server_configs.count);

	for (i=0; i<main_jobs_controller->core_config.server_configs.count;i++)
	{
		mashup_server_endpoint_t *endpoint = &(main_jobs_controller->core_config.server_configs.el[i]);

		// creamos server_socket.
		if ((endpoint->server_socket = socket_init(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == NULL)
		{
			logger_core_error("%s","[jobs_controller] imposible crear socket AF_INET IPPROTO_TCP");
			ret = 1;
			break;
		}

		// abrimos el puerto TCP del server_socket
		ret=socket_listen(endpoint->server_socket,
				endpoint->listen_ip,
				endpoint->listen_port,
				endpoint->tcp_backlog);

		if (ret != 0)
		{
			logger_core_error("[jobs_controller] imposible escuchar en IP='%s' PORT=%d BACKLOG_QUEUE=%d. Error=%d", endpoint->listen_ip, endpoint->listen_port, endpoint->tcp_backlog,ret);
			ret = 1;
			break;
		}
		logger_core_info("[jobs_controller] escuchando en IP='%s' PORT=%d BACKLOG_QUEUE=%d", endpoint->listen_ip, endpoint->listen_port, endpoint->tcp_backlog);

		// Configuro el interes permanente de ser notificado para nuevas conexiones que lleguen en server_socket.
		endpoint->server_event_config.source = JOB_SOURCE_SOCKET;
		endpoint->server_event_config.job = NULL; // Usamos un job NULL para identificar la tarea de jobs_controller.
		endpoint->server_event_config.data.sock_conf.st=(endpoint->server_socket);
		endpoint->server_event_config.operation=JOB_EVENT_OP_ADD;
		endpoint->server_event_config.asked_events=JOB_EVENT_INPUT;

		/*
			Server socket se configura como level_triggered porque queremos ser notificados cada vez que llega una conexion de
			un cliente. Si Multiaccept = 0 entonces usamos edge_triggered, ya que desde jobs_controller se procede a aceptar
			conexiones hasta que se detecta que no hay mas conexiones pendientes.
		*/
		endpoint->server_event_config.persistent=1;
		endpoint->server_event_config.edge_or_level=((endpoint->multiaccept == 0)?0:1);

		ret = io_handler_register_event_interest(main_jobs_controller->ioh,&(endpoint->server_event_config));
		if (ret != 0)
		{
			logger_core_error("[jobs_controller] error registrando eventos para server de IP='%s' PORT=%d BACKLOG_QUEUE=%d. Error=%d", endpoint->listen_ip, endpoint->listen_port, endpoint->tcp_backlog,ret);
			break;
		}
	}
	return ret;
}

static inline int jobs_controller_append_job(server_worker_job_t job, int pending_events_count)
{
	int commit=0;
	int ret=0;
	int pushed=0;

	// asumiendo que MAX_PENDING_JOBS_COMMIT > 1, nunca llegare a esta funcion con el buffer de pending jobs lleno.

	// si hay mas eventos pendientes --> acumulo. Si luego se llena --> commit.
	// si no hay mas elementos pendientes, agrego y comiteo lo acumulado.

	if (pending_events_count > 0)
	{
		main_jobs_controller->p_jobs.worker_jobs[main_jobs_controller->p_jobs.count]=job;
		main_jobs_controller->p_jobs.count++;
		if (main_jobs_controller->p_jobs.count == MAX_PENDING_JOBS_COMMIT)
		{
			commit=1;
		}
	}
	else
	{
		// agrego y comiteo.
		main_jobs_controller->p_jobs.worker_jobs[main_jobs_controller->p_jobs.count]=job;
		main_jobs_controller->p_jobs.count++;
		commit=1;
	}
	if (commit == 1)
	{
			pushed=worker_push_job_vector(main_jobs_controller->worker, (void **)(main_jobs_controller->p_jobs.worker_jobs), main_jobs_controller->p_jobs.count);
			ret= (pushed == main_jobs_controller->p_jobs.count)?0:1;
			main_jobs_controller->p_jobs.count=0;
	}
	return ret;
}

// Si es un evento asociado a un socket servidor, tenemos un evento de socket y el job es null
static inline int jobs_controller_is_server_event(job_event_t event)
{
	return (event->source == JOB_SOURCE_SOCKET) && (event->job == NULL);
}

static int jobs_controller_event_fired(job_event_t event, int pending_events_count)
{
	int ret=0;
	int iter_multiaccept=0;
	static int max_multiaccept=0;
	socket_t client_socket = NULL;
	mashup_server_endpoint_t *endpoint = NULL;


	if (jobs_controller_is_server_event(event))
	{
		if (event->fired_events & JOB_EVENT_INPUT)
		{
			// recuperamos el endpoint configurado.
			endpoint = jobs_controller_get_server_endpoint(event);
			//logger_core_error("[jobs_controller] ubico endpoint en puerto %d", endpoint->listen_port);

			for (iter_multiaccept=0; ; iter_multiaccept++)
			{
				// Accept de una nueva conexion.
				client_socket = socket_accept(endpoint->server_socket);
				if (client_socket == NULL)
				{
					// no hay mas conexiones para aceptar.
					break;
				}
				else
				{
					// Creamos nueva tarea asociada a la nueva conexion aceptada
					server_worker_job_t worker_job;
				
					worker_job = server_worker_job_init();
				
					// inicializacion de la tarea.
					server_worker_job_set_proc(worker_job,main_jobs_controller->impl.fn_app_implementation);
				
					// asigno la nueva tarea a la conexion (client socket).
					socket_set_server_worker_job(client_socket, worker_job);
				
					// pasamos los parametros a la tarea (en este caso, pasamos el socket del cliente).
					server_worker_job_set_args(worker_job,(void*)client_socket);

					// agregamos el job a la lista de tareas a enviar a la cola.
					jobs_controller_append_job(worker_job, pending_events_count);
				}
				if ((endpoint->multiaccept > 0) && (iter_multiaccept == (endpoint->multiaccept-1)))
				{
					break;
				}
			}
			if (max_multiaccept < iter_multiaccept)
			{
				max_multiaccept=iter_multiaccept;
				logger_core_error("[jobs_controller] multiaccept max  = %d. Pending: %d", max_multiaccept, pending_events_count);
			}
		}
		else
		{
			logger_core_error("[jobs_controller] eventos disparados: '%d' en server_socket!", event->fired_events);
		}
	}
	else
	{
		// se trata de una conexion previamente establecida que genero un evento.
		
		// Asigno el estado a la tarea "listo para ser ejecutada".
		server_worker_job_set_status(event->job,SERVER_WORKER_JOB_RUNNABLE);

		// agregamos el job a la lista de tareas a enviar a la cola.		
		jobs_controller_append_job(event->job, pending_events_count);

	}
	return ret;
}

// funcion principal de este hilo: esperar eventos del modulo io_handler.
static int jobs_controller_main(void)
{
	int ret=0;
	// Esperamos los eventos en el modulo io_handler.
	while (1)
	{
		// reseteo la cantidad de eventos pendientes de procesar.
		main_jobs_controller->p_jobs.count=0;

		// esperamos eventos.
		ret=io_handler_process_events(main_jobs_controller->ioh);

	}
	return ret;
}

// destruccion de jobs_controller.
int jobs_controller_destroy()
{
	int i=0;
	
	if (main_jobs_controller == NULL) return 1;
	if (main_jobs_controller->worker != NULL)
	{
		worker_destroy(main_jobs_controller->worker);
		main_jobs_controller->worker=NULL;
	}	
	if (main_jobs_controller->server_workers != NULL)
	{
		for (i=0; i<main_jobs_controller->core_config.workers ; i++)
		{
			if (main_jobs_controller->server_workers[i] != NULL)
			{
				server_worker_destroy(main_jobs_controller->server_workers[i]);
			}
		}
		free(main_jobs_controller->server_workers);
		main_jobs_controller->server_workers=NULL;
	}
	if (main_jobs_controller->q != NULL)
	{
		priority_queue_destroy(main_jobs_controller->q);
		main_jobs_controller->q=NULL;	
	}
	if (main_jobs_controller->ioh != NULL)
	{
		io_handler_destroy(main_jobs_controller->ioh);
		main_jobs_controller->ioh=NULL;	
	}
	
	// liberamos todos los puertos servidor.
	jobs_controller_endpoints_destroy();

	common_objects_pool_destroy();

	if (main_jobs_controller->impl.fn_app_destroy != NULL)
	{
		main_jobs_controller->impl.fn_app_destroy();
	}
	worker_destroy_module();
	

	if (main_jobs_controller->impl.lib_handle != NULL)
	{
		dlclose(main_jobs_controller->impl.lib_handle);
		main_jobs_controller->impl.lib_handle=NULL;
	}
	main_jobs_controller=NULL;

	return 0;
}

// cargo la aplicacion que mashup-server ejecutara.
static int jobs_controller_load_server_implementation()
{
	int ret=0;
	main_jobs_controller->impl.lib_handle = dlopen(getenv("MASHUP_SERVER_IMPL"), RTLD_NOW);
	if (main_jobs_controller->impl.lib_handle == (void *)NULL) {
		logger_core_error("[jobs_controller] error cargando libreria dinamica '%s'. %s", getenv("MASHUP_SERVER_IMPL"),dlerror());
		return 1;
	}
	ret=jobs_controller_get_application_function((void **)(&(main_jobs_controller->impl.fn_app_init)),"server_worker_implementation_init");
	if (ret == 0)
	{
		ret=jobs_controller_get_application_function((void **)(&(main_jobs_controller->impl.fn_app_destroy)),"server_worker_implementation_destroy");
	}
	if (ret == 0)
	{
		ret=jobs_controller_get_application_function((void **)(&(main_jobs_controller->impl.fn_app_implementation)),"server_worker_implementation");
	}
	return ret;
}

static int jobs_controller_init_server_implementation()
{
	// Inicializo el modulo de la aplicacion.
	if (main_jobs_controller->impl.fn_app_init != NULL)
	{
		return main_jobs_controller->impl.fn_app_init();
	}
	return 1;
}

/**
	inicializacion de jobs_controller. Retorna 0 si ok, sino distinto de 0.
**/
int jobs_controller_init()
{
		int i=0, ret=0;
		server_worker_t wk = NULL;
		//object_pool_conf pool_conf[10];
		
		main_jobs_controller->server_workers=NULL;
		main_jobs_controller->worker=NULL;
		main_jobs_controller->ioh = NULL;
		main_jobs_controller->q = NULL;

		if ((ret=jobs_controller_load_server_implementation()) != 0)
		{
			logger_core_error("[jobs_controller] imposible obtener server implementation. ret=%d", ret);
			jobs_controller_destroy();
			return 1;
		}

		/** Vemos como se lee la configuracion. **/
		if (conf_file_parse(&(main_jobs_controller->core_config), getenv("MASHUP_SERVER_CONF_FILE")) != EXIT_SUCCESS)
		{
			logger_core_error("%s","[jobs_controller] error al parsear archivo de config.");
			jobs_controller_destroy();
			return 1;
		}

		/** Inicializamos los pools. **/
		ret=common_objects_pool_init(main_jobs_controller->core_config.pool_conf.count,main_jobs_controller->core_config.pool_conf.el);
		if (ret != 0)
		{
			logger_core_error("[jobs_controller] imposible crear common_objects_pool. ret=%d", ret);
			jobs_controller_destroy();
			return 1;
		}

		// inicializacion del modulo worker.
		if ((ret=worker_init_module()) != 0)
		{
			logger_core_error("[jobs_controller] error inicializando modulo worker. ret=%d", ret);
			jobs_controller_destroy();
			return 1;
		}

		if ((ret=jobs_controller_init_server_implementation()) != 0)
		{
			logger_core_error("[jobs_controller] imposible inicializar server implementation. ret=%d", ret);
			jobs_controller_destroy();
			return 1;
		}
			
		main_jobs_controller->server_workers=(server_worker_t*)malloc(main_jobs_controller->core_config.workers*sizeof(server_worker_t));
		if (main_jobs_controller->server_workers == NULL)
		{
			logger_core_error("%s","[jobs_controller] imposible inicializar instancias de workers");
			jobs_controller_destroy();
			return 1;
		}
		
		if ((main_jobs_controller->ioh = io_handler_init_max_events(1024, jobs_controller_event_fired)) == NULL)
		{
			logger_core_error("%s","[jobs_controller] imposible inicializar io_handler");
			jobs_controller_destroy();
			return 1;
		}
		
		main_jobs_controller->q=priority_queue_create();
		if (main_jobs_controller->q == NULL)
		{
			logger_core_error("%s","[jobs_controller] imposible crear cola de tareas");
			jobs_controller_destroy();
			return 1;
		}
		
		jobs_controller_info_priority();

		//inicializo la cola de prioridad.
		if (priority_queue_init(main_jobs_controller->q,main_jobs_controller->core_config.q_conf) != 0)
		{
			logger_core_error("%s","[jobs_controller] imposible inicializar cola de tareas. Verificar parametros de prioridad.");
			jobs_controller_destroy();
			return 1;
		}


		// Inicializo el vector de server_workers (los hilos de ejecucion que procesan los requerimientos).
		for (i=0; i<main_jobs_controller->core_config.workers;i++)
		{
			main_jobs_controller->server_workers[i]=NULL;
		}
		for (i=0; i<main_jobs_controller->core_config.workers;i++)
		{
			wk=server_worker_init(main_jobs_controller->q, main_jobs_controller, i+1);
			logger_core_info("%s %d", "[jobs_controller] creando worker", i);
			if (wk == NULL)
			{
				logger_core_error("%s %d", "[jobs_controller] imposible crear worker", i);
				jobs_controller_destroy();
				return 1;
			}
			else
			{
				main_jobs_controller->server_workers[i]=wk;
			}
		}
		
		
		/** Inicializo la tarea de jobs_controller. **/
		main_jobs_controller->worker=worker_init((void *)jobs_controller_main, NULL, main_jobs_controller->q,64*K, 0);
		if (main_jobs_controller->worker==NULL)
		{
			logger_core_error("%s","[jobs_controller] imposible crear worker para jobs_controller");
			jobs_controller_destroy();
			return 1;
		}
		ret=worker_start(main_jobs_controller->worker);
		if (ret != 0)
		{
			logger_core_error("%s","[jobs_controller] imposible iniciar worker para jobs_controller");
			jobs_controller_destroy();
			return 1;
		}
		
		/** Abrimos todos los puertos servidores para que inicie la aplicación. **/
		ret=jobs_controller_start_listeners();
		if (ret != 0)
		{
			logger_core_error("[jobs_controller] error iniciando puertos servidor: %d",ret);
			jobs_controller_destroy();
			return 1;
		}
		return 0;
}

int jobs_controller_register_event_interest(job_event_t event)
{
	return io_handler_register_event_interest(main_jobs_controller->ioh, event);
}

// forma en que una aplicacion tiene de lanzar un job a pedido.
int jobs_controller_launch_child(server_worker_job_t job)
{
	return worker_push_job_priority(main_jobs_controller->worker, job, server_worker_job_get_launch_priority(job));
}
